package drds.global_transaction.api.rpc.rpc_client;

import java.net.InetSocketAddress;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.commons.pool.impl.GenericKeyedObjectPool;
import org.apache.commons.pool.impl.GenericKeyedObjectPool.Config;

import drds.global_transaction.api.message_codec.HeartbeatMessage;
import drds.global_transaction.api.message_codec.MergeResponseMessage;
import drds.global_transaction.api.message_codec.MergeWarpMessage;
import drds.global_transaction.api.message_codec.rpc_message.RpcMessage;
import drds.global_transaction.api.message_codec.rpc_message.RpcMessageFuture;
import drds.global_transaction.api.rpc.AbstractRpcChannelDuplexHandler;
import drds.global_transaction.api.rpc.MessageCodecHandler;
import drds.global_transaction.api.rpc.RpcService;
import drds.global_transaction.api.rpc.config.ClientConfig;
import drds.global_transaction.api.rpc.rpc_client.resource_manager.IResourceManagerClientMessageListener;
import drds.global_transaction.api.rpc.rpc_client.server_addresss_look_up_service.GlobalTransactionServiceGroupAddressManager;
import drds.global_transaction.api.rpc.rpc_client.server_addresss_look_up_service.GlobalTransactionServiceGroupAddressManagerStaticConfigImpl;
import drds.global_transaction.common.exception.FrameworkException;
import drds.global_transaction.common.thread.NamedThreadFactory;
import drds.global_transaction.common.util.Nets;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.pool.AbstractChannelPoolMap;
import io.netty.channel.pool.ChannelHealthChecker;
import io.netty.channel.pool.FixedChannelPool;
import io.netty.channel.pool.FixedChannelPool.AcquireTimeoutAction;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import io.netty.util.concurrent.EventExecutorGroup;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public abstract class RpcClient extends AbstractRpcChannelDuplexHandler implements RpcService, RegisterMessageListener, ClientMessageSender
{
	public static final String msg_id_prefix = "message_id:";
	public static final String futures_prefix = "rpc_message_id_to_rpc_message_future_map:";
	public static final String single_log_postfix = ";";
	public static final int max_merge_send_mills = 1;
	public static final String thread_prefix_split_char = "_";
	@Setter
	@Getter
	protected GlobalTransactionServiceGroupAddressManager globalTransactionServiceGroupAddressManager;
	@Setter
	@Getter
	protected GenericKeyedObjectPool<Key, Channel> keyToChannelMap;
	@Setter
	@Getter
	protected IResourceManagerClientMessageListener resourceManagerClientMessageListener;
	//
	@Setter
	@Getter
	private ClientConfig clientConfig;
	@Setter
	@Getter
	private Bootstrap bootstrap = new Bootstrap();
	@Setter
	@Getter
	private EventLoopGroup eventLoopGroup;
	@Setter
	@Getter
	private AtomicBoolean initialized = new AtomicBoolean(false);
	@Setter
	@Getter
	private EventExecutorGroup eventExecutorGroup;
	@Setter
	@Getter
	private AbstractChannelPoolMap<InetSocketAddress, FixedChannelPool> inetSocketAddressToFixedChannelPoolMap;

	//
	public RpcClient(ClientConfig clientConfig, final EventExecutorGroup eventExecutorGroup, final ThreadPoolExecutor threadPoolExecutor)
	{
		super(threadPoolExecutor);
		//
		if (null == clientConfig)
		{
			clientConfig = new ClientConfig();
			if (log.isInfoEnabled())
			{
				log.info("use default netty client config.");
			}
		}
		//
		this.clientConfig = clientConfig;
		int selectorThreadSizeThreadSize = this.clientConfig.getClientSelectorThreadSize();
		this.eventLoopGroup = new NioEventLoopGroup(selectorThreadSizeThreadSize, new NamedThreadFactory(getThreadPrefix(this.clientConfig.getClientSelectorThreadPrefix()), selectorThreadSizeThreadSize));
		this.eventExecutorGroup = eventExecutorGroup;
	}

	public RpcClient(final ClientConfig clientConfig)
	{
		this(clientConfig, null, null);
	}

	//
	@Override
	public void init()
	{
		AddressToChannelFactory addressToChannelFactory = new AddressToChannelFactory(this);
		keyToChannelMap = new GenericKeyedObjectPool(addressToChannelFactory);
		keyToChannelMap.setConfig(getConfig());
		globalTransactionServiceGroupAddressManager = new GlobalTransactionServiceGroupAddressManagerStaticConfigImpl();
		super.init();
	}

	@Override
	public void start()
	{

		if (this.eventExecutorGroup == null)
		{
			this.eventExecutorGroup = new DefaultEventExecutorGroup(clientConfig.getClientWorkerThreads(), new NamedThreadFactory(getThreadPrefix(clientConfig.getClientWorkerThreadPrefix()), clientConfig.getClientWorkerThreads()));
		}
		this.bootstrap.group(this.eventLoopGroup).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true).option(ChannelOption.SO_KEEPALIVE, true).option(ChannelOption.CONNECT_TIMEOUT_MILLIS, clientConfig.getConnectTimeoutMillis()).option(ChannelOption.SO_SNDBUF, clientConfig.getClientSocketSndBufSize()).option(ChannelOption.SO_RCVBUF, clientConfig.getClientSocketRcvBufSize());

		if (clientConfig.isUseConnPool())
		{
			inetSocketAddressToFixedChannelPoolMap = new AbstractChannelPoolMap<InetSocketAddress, FixedChannelPool>()
			{
				@Override
				protected FixedChannelPool newPool(InetSocketAddress inetSocketAddress)
				{
					FixedChannelPool fixedChannelPool = new FixedChannelPool(bootstrap.remoteAddress(inetSocketAddress), new ChannelPoolHandler()
					{
						@Override
						public void channelCreated(Channel channel) throws Exception
						{
							super.channelCreated(channel);
							final ChannelPipeline channelPipeline = channel.pipeline();
							channelPipeline.addLast(eventExecutorGroup, //
									new IdleStateHandler(clientConfig.getChannelMaxReadIdleSeconds(), clientConfig.getChannelMaxWriteIdleSeconds(), clientConfig.getChannelMaxAllIdleSeconds()));
							//
							channelPipeline.addLast(eventExecutorGroup, new RpcClientHandler());// 真正的业务处理器
						}
					}, ChannelHealthChecker.ACTIVE, AcquireTimeoutAction.FAIL, clientConfig.getMaxAcquireConnMills(), clientConfig.getPerHostMaxConn(), clientConfig.getPendingConnSize(), false);
					return fixedChannelPool;

				}
			};
		} else
		{
			bootstrap.handler(new ChannelInitializer<SocketChannel>()
			{

				@Override
				public void initChannel(SocketChannel socketChannel)
				{
					ChannelPipeline channelPipeline = socketChannel.pipeline();
					channelPipeline.addLast(new IdleStateHandler(clientConfig.getChannelMaxReadIdleSeconds(), clientConfig.getChannelMaxWriteIdleSeconds(), clientConfig.getChannelMaxAllIdleSeconds())).addLast(new MessageCodecHandler());
					if (null != channelHandlers)
					{
						addChannelHandlersToChannelPipelineLast(socketChannel, channelHandlers);
					}
				}
			});
		}
		if (initialized.compareAndSet(false, true) && log.isInfoEnabled())
		{
			log.info("RpcClient has started");
		}
	}

	/**
	 * 核心方法 真正连接远程的服务器
	 */
	public Channel connect(InetSocketAddress inetSocketAddress)
	{
		ChannelFuture channelFuture = this.bootstrap.connect(inetSocketAddress);
		try
		{
			channelFuture.await(this.clientConfig.getConnectTimeoutMillis(), TimeUnit.MILLISECONDS);
			if (channelFuture.isCancelled())
			{
				throw new FrameworkException("connect cancelled, can not connect to transaction_coordinator.");
			} else if (!channelFuture.isSuccess())
			{
				throw new FrameworkException("connect failed, can not connect to transaction_coordinator.");
			} else
			{
				return channelFuture.channel();
			}
		} catch (Exception e)
		{
			throw new FrameworkException(e, "can not connect to transaction_coordinator.");
		}
	}

	@Override
	public void shutdown()
	{
		try
		{
			if (null != inetSocketAddressToFixedChannelPoolMap)
			{
				inetSocketAddressToFixedChannelPoolMap.close();
			}
			this.eventLoopGroup.shutdownGracefully();
			if (this.eventExecutorGroup != null)
			{
				this.eventExecutorGroup.shutdownGracefully();
			}
			super.destroy();
		} catch (Exception exx)
		{
			log.error("shutdown error:" + exx.getMessage());
		}
	}

	@Override
	public void channelRead(final ChannelHandlerContext channelHandlerContext, Object message) throws Exception
	{
		if (message instanceof RpcMessage)
		{
			RpcMessage rpcMessage = (RpcMessage) message;
			if (rpcMessage.getMessageCodec() == HeartbeatMessage.pong)
			{
				if (log.isDebugEnabled())
				{
					log.debug("received pong from " + channelHandlerContext.channel().remoteAddress());
				}
				return;
			}
		}

		if (((RpcMessage) message).getMessageCodec() instanceof MergeResponseMessage)
		{
			MergeResponseMessage mergeResponseMessage = (MergeResponseMessage) ((RpcMessage) message).getMessageCodec();
			MergeWarpMessage mergedWarpMessage = (MergeWarpMessage) messageIdToMergedWarpMessageMap.remove(((RpcMessage) message).getId());
			int size = mergedWarpMessage.messageList.size();
			for (int i = 0; i < size; i++)
			{
				long rpcMessageId = mergedWarpMessage.rpcMessageIdList.get(i);
				RpcMessageFuture rpcMessageFuture = rpcMessageIdToRpcMessageFutureMap.remove(rpcMessageId);
				if (rpcMessageFuture == null)
				{
					if (log.isInfoEnabled())
					{
						log.info("msg:" + rpcMessageId + " is not found in rpcMessageIdToRpcMessageFutureMap.");
					}
				} else
				{
					rpcMessageFuture.setResultMessage(mergeResponseMessage.getResponseMessages()[i]);
				}
			}
			return;
		}
		super.channelRead(channelHandlerContext, message);
	}

	@Override
	public void dispatch(long messageId, ChannelHandlerContext channelHandlerContext, Object message)
	{
		if (resourceManagerClientMessageListener != null)
		{
			String hostAddressAndPort = Nets.getHostAddressAndPort(channelHandlerContext.channel().remoteAddress());
			resourceManagerClientMessageListener.onResourceManagerClientMessage(hostAddressAndPort, messageId, message, this);
		}
	}

	protected String getThreadPrefix(String threadPrefix)
	{
		return threadPrefix + thread_prefix_split_char + getRole().name();
	}

	protected abstract Channel connect(String serverAddress);

	protected abstract Config getConfig();

	protected abstract Role getRole();

}
